package cn.itcast.flink.start;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
/** Flink State 中KeyedState，默认情况下框架自己维护，此外可以手动维护 */
public class StreamKeyedStateDemo_1 {
public static void main(String[] args) throws Exception {
// 1. 执行环境-env
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1);

// 2. 数据源-source
DataStreamSource<Tuple3<String, String, Long>> tupleStream = env.fromElements(
        Tuple3.of("上海", "普陀区", 488L),
        Tuple3.of("上海", "徐汇区", 212L),
        Tuple3.of("北京", "西城区", 823L),
        Tuple3.of("北京", "海淀区", 234L),
        Tuple3.of("上海", "杨浦区", 888L),
        Tuple3.of("上海", "浦东新区", 666L),
        Tuple3.of("北京", "东城区", 323L),
        Tuple3.of("上海", "黄浦区", 111L)
);

// 3. 数据转换-transformation
// TODO： 直 接 使 用 maxBy 的 状 态 ，Flink 自 动 维 护 了 State
  SingleOutputStreamOperator<Tuple3<String, String, Long>> maxOperator = tupleStream
.keyBy(0)
.maxBy(2);
/*
maxBy> (上海,普陀区,488) maxBy> (上海,普陀区,488) maxBy> (北京,西城区,823) maxBy> (北京,西城区,823) maxBy> (上海,杨浦区,888) maxBy> (上海,杨浦区,888) maxBy> (北京,西城区,823) maxBy> (上海,杨浦区,888)
*/
maxOperator.printToErr("maxBy");

// TODO: 使用KeyedState中的ValueState来模拟maxBy的底层,手动维护状态
SingleOutputStreamOperator<String> stateOperator = tupleStream
.keyBy(0)
.map(new RichMapFunction<Tuple3<String, String, Long>, String>() {


    // a. 定义一个状态存储最大值
private transient ValueState<Long> maxValueState;

@Override
public void open(Configuration parameters) throws Exception {
// b. 创建状态描述符
ValueStateDescriptor<Long> stateDescriptor = new ValueStateDescriptor<Long>( "maxValueState", Long.class
);
// c. 实例化状态对象
maxValueState = getRuntimeContext().getState(stateDescriptor);
}
    @Override
    public String map(Tuple3<String, String, Long> value) throws Exception {


// d. 获取历史状态
        Long historyValue = maxValueState.value();
// max 最大值比较
        Long currentValue = value.f2;
        if (null ==historyValue || currentValue >historyValue){
        // e. 更新状态
            maxValueState.update(currentValue);
            return value.f0+", "+value.f1+", "+currentValue;
        }
       else {
            return value.f0+", "+value.f1+", "+currentValue;
        }
}
});

/*
state> 上海, 普陀区, 488 state> 上海, 徐汇区, 488 state> 北京, 西城区, 823 state> 北京, 海淀区, 823 state> 上海, 杨浦区, 888 state> 上海, 浦东新区, 888 state> 北京, 东城区, 823 state> 上海, 黄浦区, 888
*/ stateOperator.print("state");
// 5. 触发执行-execute
    env.execute(StreamKeyedStateDemo_1.class.getSimpleName());
}
}
